package cn.itcast.flink.connector;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @author lilulu
 */

/**
 * 从kafka消费数据，指定topic名称和反序列化的类
 */
public class ConnectorFlinkKafkaConsumerDemo {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        // 2. 数据源-source
//        2.1 创建消费kafka数据时的属性
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        properties.setProperty("group.id", "test");
//        2.2 构建FlinkKafkaConsumer实例对象
        FlinkKafkaConsumer<String> stringFlinkKafkaConsumer = new FlinkKafkaConsumer<String>(
                "flink-topic",
                new SimpleStringSchema(),
                properties
        );
//        2.3 添加source
        DataStreamSource<String> kafkaStream = env.addSource(stringFlinkKafkaConsumer);
        // 3. 数据转换-transformation
        // 4. 数据终端-sink
        kafkaStream.print();
        // 5. 触发执行-execute
        env.execute("ConnectorFlinkKafkaConsumerDemo");
    }
}